package com.sjc.lesson02.state.keyedState.demo;


import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

public class OrderStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> info1 = env.addSource(new FileSource(Constants.ORDER_INFO1_PATH));
        DataStreamSource<String> info2 = env.addSource(new FileSource(Constants.ORDER_INFO2_PATH));

        KeyedStream<OrderInfo1, Long> orderInfo1Stream =
                info1.map(line -> OrderInfo1.string2OrderInfo1(line)).keyBy(OrderInfo1::getOrderId);

        KeyedStream<OrderInfo2, Long> orderInfo2Stream =
                info2.map(line -> OrderInfo2.string2OrderInfo2(line)).keyBy(OrderInfo2::getOrderId);

        orderInfo1Stream.connect(orderInfo2Stream)
                .flatMap(new EnrichmentFunction())
                .print();

        env.execute("OrderSteam");
    }

    /**
     * IN1,第一个流的输入的数据类型
     * IN1,第一个流的输入的数据类型
     * OUT，输出的数据类型
     */
    public static class EnrichmentFunction extends RichCoFlatMapFunction<OrderInfo1,OrderInfo2, Tuple2<OrderInfo1,OrderInfo2>>{
        // 定义第一个流 key 对应的state
        private ValueState<OrderInfo1> orderInfo1State;
        // 定义第二个流 key 对应的state
        private ValueState<OrderInfo2> orderInfo2State;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderInfo1State = getRuntimeContext().getState(new ValueStateDescriptor<OrderInfo1>("info1", OrderInfo1.class));
            orderInfo2State = getRuntimeContext().getState(new ValueStateDescriptor<OrderInfo2>("info2", OrderInfo2.class));
        }

        @Override
        public void flatMap1(OrderInfo1 orderInfo1, Collector<Tuple2<OrderInfo1, OrderInfo2>> collector) throws Exception {
            OrderInfo2 value2 = orderInfo2State.value();
            if(value2 != null){
                orderInfo2State.clear();
                collector.collect(Tuple2.of(orderInfo1,value2));
            }else{
                orderInfo1State.update(orderInfo1);
            }
        }

        @Override
        public void flatMap2(OrderInfo2 orderInfo2, Collector<Tuple2<OrderInfo1, OrderInfo2>> collector) throws Exception {
            OrderInfo1 value1 = orderInfo1State.value();
            if(value1 != null){
                orderInfo1State.clear();
                collector.collect(Tuple2.of(value1, orderInfo2));
            }else {
                orderInfo2State.update(orderInfo2);
            }
        }
    }

}
